package com.mio.flinkdemo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount {

	public static void main(String[] args) throws Exception {

		String inputPath = "/Users/mio/Desktop/word.txt";

		// 获取flink的运行环境
		ExecutionEnvironment bevt = ExecutionEnvironment.getExecutionEnvironment();

		 // 1、读取数据源
        DataSet<String> text = bevt.readTextFile(inputPath);
        //2、处理数据源
        //将读到的数据源文件中的每一行根据空格切分，将切分好的每个单词拼接1
        FlatMapOperator<String, Tuple2<String, Integer>> wordOndOnes = text.flatMap(new SplitClz());
        //根据单词聚合（将相同的单词放在一起）
        UnsortedGrouping<Tuple2<String, Integer>> groupedWordAndOne = wordOndOnes.groupBy(0);

        //累加相同的单词（单词后面的1进行累加）
        AggregateOperator<Tuple2<String, Integer>> out = groupedWordAndOne.sum(1);

        out.print();
        bevt.execute();
        
	}

	 static class SplitClz implements FlatMapFunction<String, Tuple2<String,Integer>> {

	        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
	            String[] s1 = s.split(" ");
	            for(String word : s1) {
	                collector.collect(new Tuple2<String, Integer>(word,1));
	            }
	        }
	 }
}
